package io.openkunlun.javadsl.server;

import io.grpc.Server;
import io.grpc.ServerInterceptors;
import io.grpc.netty.NettyServerBuilder;
import io.grpc.stub.StreamObserver;
import io.netty.handler.ssl.SslContext;
import io.openkunlun.javadsl.DaprExceptionResolver;
import io.openkunlun.javadsl.client.DaprClient;
import io.openkunlun.javadsl.v1.DaprProtos;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

public final class DaprServer {

    private static final int RETRY_BASE_MILLIS = 300;
    private static final int RETRY_JITTER_MILLIS = 3_000;
    private static final Logger LOG = LoggerFactory.getLogger(DaprServer.class);

    private final DaprApp app;
    private final DaprClient client;
    private final Server server;
    private final FailureDetector failureDetector;
    private final AtomicReference<Status> status = new AtomicReference<>(Status.INIT);
    private volatile long startup;

    private volatile ScheduledThreadPoolExecutor scheduled;
    private volatile ScheduledFuture<?> scheduledKeepAlive;
    private volatile ScheduledFuture<?> scheduledRetry;

    enum Status {
        UP, DOWN, INIT
    }

    DaprServer(DaprApp app, DaprClient client, Server server, FailureDetector failureDetector) {
        this.app = app;
        this.client = client;
        this.server = server;
        this.failureDetector = failureDetector;
    }

    /**
     * @param client
     * @param port
     * @return
     */
    public static Builder forPort(DaprClient client, int port) {
        return new Builder(client, port);
    }

    /**
     * @param client
     * @param host
     * @param port
     * @return
     */
    public static Builder forAddress(DaprClient client, String host, int port) {
        return new Builder(client, host, port);
    }

    /**
     * @return
     * @throws IOException
     */
    public DaprServer start() throws IOException {
        if (status.compareAndSet(Status.INIT, Status.UP)) {
            startInternal();
            LOG.info("service started. listen on {}:{}", app.host(), app.port());
        }

        return this;
    }

    private void startInternal() throws IOException {
        server.start();
        startup = System.currentTimeMillis();
        if (scheduled == null) {
            scheduled = new ScheduledThreadPoolExecutor(1, new ThreadFactory() {

                private final AtomicLong threadNumber = new AtomicLong(1);

                @Override
                public Thread newThread(Runnable r) {
                    return new Thread(null, r, "dapr-keepalive-" + threadNumber.getAndIncrement());
                }
            });
        }
        scheduled.submit(() -> establishApp(false));
    }

    /**
     * @return
     */
    public DaprServer stop() {
        if (status.compareAndSet(Status.UP, Status.DOWN)) {
            if (scheduledRetry != null) {
                scheduledRetry.cancel(true);
            }

            if (scheduledKeepAlive != null) {
                scheduledKeepAlive.cancel(true);
            }

            stopInternal();
            LOG.info("service stopped.");
        }

        return this;
    }

    private void stopInternal() {
        final CountDownLatch signal = new CountDownLatch(1);
        abolishApp(new StreamObserver<DaprProtos.AbolishResponse>() {
            @Override
            public void onNext(DaprProtos.AbolishResponse abolishResponse) {
            }

            @Override
            public void onError(Throwable ignore) {
                LOG.error("Abolish app failure. it may cause an inconsistent window for requests.");
                shutdown();
            }

            @Override
            public void onCompleted() {
                shutdown();
            }

            private void shutdown() {
                try {
                    if (scheduled != null) {
                        scheduled.shutdown();
                    }
                } catch (Exception ignore) {
                }
                try {
                    server.shutdown();
                } catch (Exception e) {
                    // ignore
                }

                signal.countDown();
            }
        });

        try {
            signal.await();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    private void abolishApp(StreamObserver<DaprProtos.AbolishResponse> responseObserver) {
        DaprProtos.AbolishRequest request = DaprProtos.AbolishRequest.newBuilder().setId(app.id()).build();
        client.abolishService(request, responseObserver);
    }

    private void establishApp(boolean reconnect) {
        DaprProtos.EstablishRequest request = DaprProtos.EstablishRequest.newBuilder()
                .setId(app.id())
                .setHost(app.host())
                .setPort(app.port())
                .setTtl(app.ttl())
                .setWeight(app.weight())
                .setWarmup(app.warmup())
                .setStartup(startup)
                .addAllRoles(app.roles())
                .addAllBindings(app.bindings())
                .build();
        StreamObserver<DaprProtos.EstablishResponse> responseObserver = new StreamObserver<DaprProtos.EstablishResponse>() {
            @Override
            public void onNext(DaprProtos.EstablishResponse result) {
            }

            @Override
            public void onError(Throwable cause) {
                if (status.get() == Status.DOWN) {
                    return;
                }
                LOG.error("Establish app failure.", cause);
                reconnectApp();
            }

            @Override
            public void onCompleted() {
                if (status.get() == Status.DOWN) {
                    return;
                }
                keepAliveApp(request);
            }
        };

        if (reconnect) {
            // detect failure when reconnect.
            FailureDetected fd = failureDetector.detect(client);
            if (fd.isUnavailable()) {
                responseObserver.onError(fd.cause());
            } else {
                client.establishService(request, responseObserver);
            }
        } else {
            client.establishService(request, responseObserver);
        }
    }

    /**
     * @param request
     * @return
     */
    private void keepAliveApp(DaprProtos.EstablishRequest request) {
        // todo
        StreamObserver<DaprProtos.KeepAliveResponse> responseObserver = new StreamObserver<DaprProtos.KeepAliveResponse>() {
            @Override
            public void onNext(DaprProtos.KeepAliveResponse result) {
            }

            @Override
            public void onError(Throwable cause) {
                if (status.get() == Status.DOWN) {
                    return;
                }
                LOG.error("Keepalive app failure.", cause);
                reconnectApp();
            }

            @Override
            public void onCompleted() {
                if (status.get() == Status.DOWN) {
                    return;
                }
                reconnectApp();
            }
        };

        StreamObserver<DaprProtos.KeepAliveRequest> requestObserver = client.keepAliveService(responseObserver);
        DaprProtos.KeepAliveRequest keepAliveRequest = DaprProtos.KeepAliveRequest.newBuilder().setId(request.getId()).build();

        // todo
        long ttlInMills = app.ttl() * 1000L;
        long ttlPeriod = ttlInMills / 3;
        if (ttlPeriod <= 0) {
            ttlPeriod = ttlInMills / 2;
        }
        Runnable ttlTask = () -> {
            FailureDetected fd = failureDetector.detect(client);
            if (fd.isAvailable()) {
                requestObserver.onNext(keepAliveRequest);
            } else {
                responseObserver.onError(fd.cause());
            }
        };

        if (scheduledKeepAlive != null) {
            scheduledKeepAlive.cancel(true);
        }
        scheduledKeepAlive = scheduled.scheduleAtFixedRate(ttlTask, ttlPeriod, ttlPeriod, TimeUnit.MILLISECONDS);
    }

    private void reconnectApp() {

        if (status.get() == Status.DOWN) {
            return;
        }

        if (scheduledKeepAlive != null) {
            scheduledKeepAlive.cancel(true);
            scheduledKeepAlive = null;
        }

        long delay = RETRY_BASE_MILLIS + ThreadLocalRandom.current().nextLong(RETRY_JITTER_MILLIS);
        Runnable command = () -> abolishApp(new StreamObserver<DaprProtos.AbolishResponse>() {
            @Override
            public void onNext(DaprProtos.AbolishResponse result) {
            }

            @Override
            public void onError(Throwable cause) {
                reconnectApp();
            }

            @Override
            public void onCompleted() {
                establishApp(true);
            }
        });

        if (scheduledRetry != null) {
            scheduledRetry.cancel(true);
        }
        scheduledRetry = scheduled.schedule(command, delay, TimeUnit.MILLISECONDS);
    }

    public static class Builder {

        private final DaprClient client;
        private final DaprApp.Builder appBuilder;
        private final NettyServerBuilder serverBuilder;
        private FailureDetector failureDetector;

        Builder(DaprClient client, int port) {
            this(client, "127.0.0.1", port);
        }

        Builder(DaprClient client, String host, int port) {
            this.client = client;
            this.appBuilder = new DaprApp.Builder(host, port);
            this.serverBuilder = NettyServerBuilder.forAddress(new InetSocketAddress(host, port));
        }

        public Builder setId(String id) {
            appBuilder.setId(id);
            return this;
        }

        public Builder setTtl(int ttl) {
            appBuilder.setTtl(ttl);
            return this;
        }

        public Builder setWeight(int weight) {
            appBuilder.setWeight(weight);
            return this;
        }

        public Builder setWarmup(int warmup) {
            appBuilder.setWarmup(warmup);
            return this;
        }

        public Builder failureDetector(FailureDetector failureDetector) {
            this.failureDetector = failureDetector;
            return this;
        }

        public Builder addHandler(DaprHandler handler) {
            appBuilder.addHandler(handler);
            return this;
        }

        public Builder addHandlers(DaprHandler... handlers) {
            appBuilder.addHandlers(handlers);
            return this;
        }

        public Builder addHandler(BindingHandler handler) {
            appBuilder.addHandler(handler);
            return this;
        }

        public Builder addHandlers(BindingHandler... handlers) {
            appBuilder.addHandlers(handlers);
            return this;
        }

        public Builder addHandler(InvocationHandler handler) {
            appBuilder.addHandler(handler);
            return this;
        }

        public Builder addHandlers(InvocationHandler... handlers) {
            appBuilder.addHandlers(handlers);
            return this;
        }

        public Builder errorHandler(DaprExceptionResolver errorHandler) {
            appBuilder.errorHandler(errorHandler);
            return this;
        }

        public Builder sslContext(SslContext sslContext) {
            serverBuilder.sslContext(sslContext);
            return this;
        }

        public Builder maxConcurrentCallsPerConnection(int maxCalls) {
            serverBuilder.maxConcurrentCallsPerConnection(maxCalls);
            return this;
        }

        public Builder maxInboundMessageSize(int bytes) {
            serverBuilder.maxInboundMessageSize(bytes);
            return this;
        }

        public Builder maxInboundMetadataSize(int bytes) {
            serverBuilder.maxInboundMetadataSize(bytes);
            return this;
        }

        public Builder keepAliveTime(long keepAliveTime, TimeUnit timeUnit) {
            serverBuilder.keepAliveTime(keepAliveTime, timeUnit);
            return this;
        }

        public Builder keepAliveTimeout(long keepAliveTimeout, TimeUnit timeUnit) {
            serverBuilder.keepAliveTimeout(keepAliveTimeout, timeUnit);
            return this;
        }

        public Builder maxConnectionIdle(long maxConnectionIdle, TimeUnit timeUnit) {
            serverBuilder.maxConnectionIdle(maxConnectionIdle, timeUnit);
            return this;
        }

        public Builder maxConnectionAge(long maxConnectionAge, TimeUnit timeUnit) {
            serverBuilder.maxConnectionAge(maxConnectionAge, timeUnit);
            return this;
        }

        public Builder maxConnectionAgeGrace(long maxConnectionAgeGrace, TimeUnit timeUnit) {
            serverBuilder.maxConnectionAgeGrace(maxConnectionAgeGrace, timeUnit);
            return this;
        }

        public Builder directExecutor() {
            serverBuilder.directExecutor();
            return this;
        }

        public Builder executor(Executor executor) {
            serverBuilder.executor(executor);
            return this;
        }

        public DaprServer build() {
            DaprApp app = appBuilder.build();
            Server server = serverBuilder.addService(ServerInterceptors.intercept(app, new DaprServerContextInterceptor())).build();
            if (failureDetector == null) {
                failureDetector = FailureDetector.AVAILABLE;
            }
            return new DaprServer(app, client, server, failureDetector);
        }
    }

}
